package com.bw.app.dwd1;

import com.alibaba.fastjson.JSONObject;
import com.bw.utils.MyKafkaUtil;
import com.bw.utils.MysqlUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;


/**
 * @projectName: Flinkgd
 * @fileName: dwdTradeCardAdd
 * @packageName: com.bw.app.dwd1
 * @date: 2025/9/23 8:41
 * @author: ymw
 */
public class dwdTradeCardAdd {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.executeSql("create table traffic_action_log(\n" +
                "    `common` map<string,string>,\n" +
                "    `action` map<string,string>,\n" +
                "    `page` map<string,string>,\n" +
                "    ts  string\n" +
                " )"+ MyKafkaUtil.getKafkaDDL("dwd_traffic_action_log","dwd_traffic_action_log1"));
        tenv.executeSql("create table base_dic(\n" +
                "    dic_code string,\n" +
                "    dic_name string,\n" +
                "    parent_code string,\n" +
                "    create_time string,\n" +
                "    operate_time string,\n" +
                "    primary key(`dic_code`) not enforced\n" +
                ")" + MysqlUtil.mysqlLookUpTableDDL("base_dic"));
//      tenv.sqlQuery("select * from base_dic").execute().print();
        Table resulttable = tenv.sqlQuery("select\n" +
                "  common['ar'] as ar,\n" +
                "  common['uid'] as uid,\n" +
                "  common['os'] as os,\n" +
                "  common['ch'] as ar,\n" +
                "  common['is_new'] as ar,\n" +
                "  common['mid'] as mid,\n" +
                "  common['vc'] as vc,\n" +
                "  common['ba'] as ba,\n" +
                "  page['item'] as sku_id,\n" +
                "  page['shop_id'] as shop_id,\n" +
                "  page['during_time'] as during_time,\n" +
                "  page['source_type'] as source_type,\n" +
                "   ts\n" +
                "from traffic_action_log\n" +
                "where action['action_id'] = 'cart_add'\n" +
                "and page['item'] is not null");
        tenv.createTemporaryView("result_table",resulttable);
//        tenv.sqlQuery("select * from  result_table").execute().print();
        tenv.executeSql("create table dwd_log_cart_add(\n" +
                "   ar string,\n" +
                "   uid string,\n" +
                "   os string,\n" +
                "   ch string,\n" +
                "   is_new string,\n" +
                "   mid string,\n" +
                "   vc string,\n" +
                "   ba string,\n" +
                "   sku_id string,\n" +
                "   shop_id string,\n" +
                "   during_time  string,\n" +
                "   source_type string,\n" +
                "   ts string\n" +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_log_cart_add"));
        tenv.executeSql("insert into dwd_log_cart_add select * from result_table");
        tenv.toAppendStream(resulttable, Row.class).print("====>");
        env.execute("dwdTradeCardAdd");
    }
}
